#include "doc/document_creator.h"
#include "doc/index_document.h"

namespace hawking {
namespace indexlib {

DEFINE_bool(field_check_strict, true, "true will discard msg which has invalid field");
DECLARE_string(default_ttl_col_name);

std::shared_ptr<NormalDocument> DocumentCreator::CreateNormalDocument(
    const indexlib::IndexlibMessage& item) const {
    // must has pk
    uint64_t pk;
    if (!FetchPK_(item, &pk)) {
        return nullptr;
    }
    
    std::shared_ptr<NormalDocument> doc = std::make_shared<NormalDocument>();
    doc->SetOperatorType(item.cmd());
    if (index_config_.life_cycle()) {
        int64_t dead_time = DocumentCreator::DeadTime(index_config_.life_cycle());
        auto& mutable_item = const_cast<indexlib::IndexlibMessage&>(item);
        ttl_col = mutable_item->add_columns();
        ttl_col->set_column_name(FLAGS_default_ttl_col_name);
        ttl_col->mutable_column_value()->set_int_value(dead_time);
    }

    std::shared_ptr<IndexDocument> index_doc =std::make_shared<IndexDocument>();
    doc->SetIndexDocument(index_doc);

    std::shared_ptr<AttributeDocument> attr_doc = std::make_shared<AttributeDocument>();
    attr_doc->Reserve(index_config_->fields_size());
    doc->SetAttributeDocument(attr_doc);

    SetPrimaryKey(pk, doc);
    if (!SetDocumentValue(item, doc)) {
        return nullptr;
    }
    return doc;
}

bool DocumentCreator::FetchPK_(
    const indexlib::IndexlibMessage& item, uint64_t* pk) const {
    for (int32_t idx = 0; idx < item.columns_size(); ++ idx) {
        if (item.columns(idx).column_name() == pk_field_name_) {
            *pk = item.columns(idx).column_value().int_value();
            return true;
        }
    }
    return false;
}

bool DocumentCreator::SetPrimaryKey(uint64_t pk, std::shared_ptr<NormalDocument> doc) {
    std::shared_ptr<IndexDocument> index_doc = doc->GetIndexDocument();
    index_doc->SetPrimaryKey(pk);
    return true;
}

bool DocumentCreator::IsIndex_(int32_t field_id) const {
    for (const auto& inverted_config : index_config_->inverted_configs()) {
        if (field_id == inverted_config.field_id()) {
            return true;
        }
    }

    return false;
}

bool DocumentCreator::SetDocumentValue(
    const indexlib::IndexlibMessage& item, std::shared_ptr<NormalDocument> doc) {
    std::unordered_set<std::string> column_name_set;
    for (const auto& column : item.columns()) {
        // ignore invalid fieldname
        auto iter = field_name2id_.find(column.column_name());
        if (iter == field_name2id_.end()) {
            if (FLAGS_field_check_strict) {
                return false;
            } else {
                continue;
            }
        }
        column_name_set.emplace(column.column_name());

        int32_t field_id = iter->second;
        if (IsIndex_(field_id)) {
            bool ret = convertor_->ConvertIndexField(
                doc, index_config_->fields(field_id), &column.column_value());
            if (!ret) {
                return false;
            }
        }

        if (!convertor_->ConvertAttributeField(
                doc, field_id, index_config_->fields(field_id), &column.column_value())) {
            return false;
        }
    }

    // missing fields set default value
    for (const auto& field_config : index_config_->fields()) {
        if (column_name_set.find(field_config.field_name() != column_name_set.end())) {
            continue;
        }

        convertor_->ConvertAttributeField(doc, field_config);
    }

    return true;
}

}
}